package org.example.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * kafka消费消息
 *
 * @author youshang
 * @date 2020/12/28 15:58
 **/
public class LocalKafkaConsumer implements Runnable {


    private final KafkaConsumer<String, String> consumer;
    private ConsumerRecords<String, String> msgList;
    private String topic = null;
    private String consumerGroup = null;
    private Object targetObject = null;
    private Method method = null;

    public LocalKafkaConsumer(String topic,String consumerGroup,Object targetObject,Method method) {
        Properties props = new Properties();
        //TODO
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-123:9092,kafka-124:9092,kafka-125:9092");
//        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfiguration.getKafkaBootstrapServers());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
        //是否默认自动提交，默认 ： true
        //props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        //默认自动提交时间为 1000 默认：5000
        //props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        //从最早,未被消费的开始消费
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<>(props);

        this.topic = topic;
        this.consumerGroup = consumerGroup;
        this.targetObject = targetObject;
        this.method = method;
        this.consumer.subscribe(Arrays.asList(topic));

    }

    @Override
    public void run() {
        System.out.println("---------开始消费---------");
        try {
            while (true) {
                msgList = consumer.poll(Duration.ofMillis(100));
                if (null != msgList && msgList.count() > 0) {
                    for (ConsumerRecord<String, String> record : msgList) {
                        method.invoke(targetObject,record);
                    }
                } else {
                    Thread.sleep(1000);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
}
